package com.lichao.pojo;

import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.*;
import java.util.Calendar;
import java.util.Properties;

/**
 * @author: li chao
 * @Desc:
 * @create: 2024-07-11 13:49
 **/
public class MysqlSource extends RichSourceFunction<BehaviorFull> {

    private volatile boolean isRunning = true;
    private Connection connection;
    private PreparedStatement statement;
    private ResultSet resultSet;

    @Override
    public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
        super.open(parameters);
        Class.forName("com.mysql.cj.jdbc.Driver");
        Properties props = new Properties();
        props.setProperty("user", "root");
        props.setProperty("password", "123456");

        // 建立数据库连接
        connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/learning", props);
    }

    @Override
    public void run(SourceContext<BehaviorFull> ctx) throws Exception {
        while (isRunning) {
            // 准备SQL查询（这里假设你有一个查询可以定期或持续返回数据）
            String sql = "SELECT * FROM behavior_full";
            statement = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            resultSet = statement.executeQuery();
            while (resultSet.next()) {
                String clientId = resultSet.getString("clientId");
                String pageId = resultSet.getString("pageId");
                String eventId = resultSet.getString("eventId");
                String arg = resultSet.getString("arg");
                // 处理获取的数据...
                BehaviorFull event = new BehaviorFull(clientId, pageId, eventId, arg, Calendar.getInstance().getTimeInMillis());
                ctx.collect(event);
            }
            Thread.sleep(20000);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
        try {
            if (resultSet != null) {
                resultSet.close();
            }
            if (statement != null) {
                statement.close();
            }
            if (connection != null) {
                connection.close();
            }
        } catch (SQLException e) {
            // 处理异常
        }
    }
}